package com.lyrcsoft.webflux_01.reactor;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.Arrays;
import java.util.concurrent.atomic.AtomicLong;

public class ReactorDemo {
    public static void main(String[] args) {
        Mono<String> mono = Mono.just("mono");
        mono.subscribe(i -> System.out.println("1个元素（Mono）" + i));
        Mono<String> mono1 = Mono.empty();
        mono1.subscribe(i -> System.out.println("空元素（Mono）" + i));
        System.out.println("==========================简单创建==========================");
        // 简单创建
        Flux<String> flux = Flux.just("flux", "flux1", "flux2");
        flux.subscribe(i -> System.out.println("多个元素（Flux）" + i));
        Flux.fromIterable(Arrays.asList("flux", "flux1", "flux2")).subscribe(i -> System.out.println("迭代元素（Flux）" + i));
        Flux.fromStream(Arrays.stream("flux,flux1,flux2".split(","))).subscribe(i -> System.out.println("流元素（Flux）" + i));
        Flux.range(1, 5).subscribe(i -> System.out.println("从1开始，5个元素（Flux）" + i));
        Flux.fromArray(new String[]{"flux", "flux1", "flux2"}).subscribe(i -> System.out.println("数组元素（Flux）" + i));
        System.out.println("===========================================================");
        System.out.println("==========================程序创建==========================");
        System.out.println("===========================================================");
        // 程序创建
        // 生成一个函数式：public static <T, S> Flux<T> generate(Callable<S> stateSupplier, BiFunction<S, SynchronousSink<T>, S> generator)
        // stateSupplier：Callable ==> 返回一个值的函数式
        // generator：BiFunction==> 两个输入参数 一个输出
        Flux<String> generate1 = Flux.generate(() -> 0, (state, sink) -> {
            // 接收消息
            sink.next("3 x " + state + " = " + 3 * state);
            if (state == 10) {
                sink.complete();
            }
            return state + 1;
        });
        generate1.subscribe(i -> System.out.println("程序创建元素（Flux）" + i));


        Flux<String> generate2 = Flux.generate(
                AtomicLong::new,
                (state, sink) -> {
                    long i = state.getAndIncrement();
                    sink.next("3 x " + i + " = " + 3 * i);
                    if (i == 10) sink.complete();
                    return state;
                });
        generate2.subscribe(i -> System.out.println("程序创建元素（Flux）" + i));

        Flux.create(sink -> {
            sink.next("flux");
            sink.next("flux1");
            sink.next("flux2");
            sink.complete();
        }).subscribe(i -> System.out.println("程序创建元素（Flux）" + i));
    }
}
